package com.github.slprj.slake.net;

import com.github.slprj.slake.common.BitUtil;
import com.github.slprj.slake.common.SlException;
import com.github.slprj.slake.log.SlLog;
import com.github.slprj.slake.log.SlLogs;
import com.github.slprj.slake.pb.PbParser;

import java.lang.ref.WeakReference;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author YangXiaoLong
 * @date 2022/8/24 0:32
 */
public class SlChannel {
    SlLog log = SlLogs.getLogger(SlChannel.class);
    private static final int FIX_FRAME_LENGTH = 4;
    private static final int INIT_BUF_SIZE = 2048;
    /**
     * 最大8M
     */
    private static final int MAX_FRAME_SIZE = 8 * 1024 * 1024;

    private final NioStatus nioStatus;
    private final NioConfig nioConfig;
    private final WorkerParams workerParams;
    private final SocketChannel channel;
    private Peer peer;

    private WeakReference<ByteBuffer> readBufferCache;
    private ByteBuffer readBuffer;

    //read status
    private int currentReadFrameSize = -1;
    private int readBufferMark = 0;

    private final int channelIndexInWorker;
    int seq = 1;

    private final IoSubQueue subQueue;
    private boolean closed;


    public SlChannel(NioStatus nioStatus, WorkerParams workerParams, NioConfig nioConfig,
                     SocketChannel socketChannel, int channelIndexInWorker
    ) {
        this.nioStatus = nioStatus;
        this.channel = socketChannel;
        this.subQueue = new IoSubQueue(workerParams.getPool());
        this.nioConfig = nioConfig;
        this.workerParams = workerParams;
        this.channelIndexInWorker = channelIndexInWorker;
    }

    public ByteBuffer getOrCreateReadBuffer() {
        if (readBuffer != null) {
            return readBuffer;
        }
        //缓存为空
        if (readBufferCache == null) {
            return createNewBuffer();
        } else {
            ByteBuffer cached = readBufferCache.get();
            if (cached != null) {
                return cached;
            } else {
                return createNewBuffer();
            }
        }
    }

    public void afterRead(boolean running) {
        //当前为写入模式
        ByteBuffer buf = this.readBuffer;
        //position：对于写入模式，表示当前可写入数据的下标，对于读取模式，表示接下来可以读取的数据的下标
        buf.limit(buf.position());
        //buf一直被使用
        buf.position(this.readBufferMark);
        //对写而言, 可写入的剩余数量  对于读而言,可读的数量
        for (int rest = buf.remaining(); rest >= this.currentReadFrameSize; rest = buf.remaining()) {
            //第一次读取数据
            if (currentReadFrameSize == -1) {
                if (rest >= FIX_FRAME_LENGTH) {
                    //读取整个数据体的长度
                    currentReadFrameSize = buf.getInt();
                    if (currentReadFrameSize <= 0 || currentReadFrameSize > MAX_FRAME_SIZE) {
                        throw new SlException("frame too large" + currentReadFrameSize);
                    }
                    //当前读取的数据的位置
                    readBufferMark = buf.position();
                } else {
                    break;
                }
            }
            if (currentReadFrameSize > 0) {
                readFrame(buf, running);
            }
        }
        prepareForNextRead();
    }

    private void prepareForNextRead() {
        //当前buffer为写模式,为下一次读数据做准备
        //0.从哪里开始读数据
        //1.读取数据完成
        //2.buffer不够,扩增buffer的容量
        ByteBuffer buf = this.readBuffer;
        //上次结束读数据的位置
        int endIndex = buf.limit();
        //buffer总的大小
        int capacity = buf.capacity();
        //数据的大小,-1时,读取数据完成
        int frameSize = this.currentReadFrameSize;
        //上次结束读数据的位置 todo  表象看来是和endIndex相同,在什么样的情况会不同呢
        int mark = this.readBufferMark;
        //数据读取完成,则重置buffer数据
        if (frameSize == -1) {
            //刚好读取数据完成,则重置buffer
            if (mark == capacity) {
                buf.clear();
                this.readBufferMark = 0;
            } else {
                int c = endIndex - mark;
                //TODO ???
                assert c >= 0 && c < 4;
                //不足下次读取数据的长度,总长是小于等于4字节的,protocolBuffer优化
                if (capacity - endIndex < 4 - c) {
                    //将数据移动到最前面
                    for (int i = 0; i < c; i++) {
                        buf.put(i, buf.get(mark + i));
                    }
                    buf.position(c);
                    buf.limit(capacity);
                    this.readBufferMark = 0;
                } else {
                    //继续读取数据
                    buf.position(endIndex);
                    buf.limit(capacity);
                }
            }
        } else {
            //buffer剩余的存储数据小于数据的长度
            int c = endIndex - mark;
            if (capacity - endIndex >= frameSize - c) {
                buf.position(endIndex);
                buf.limit(capacity);
            } else {
                buf.limit(endIndex);
                buf.position(mark);
                //空间不足
                if (frameSize > capacity) {
                    ByteBuffer nBuffer = ByteBuffer.allocateDirect(frameSize);
                    nBuffer.put(buf);
                    this.readBuffer = nBuffer;
                    this.readBufferCache = new WeakReference<>(nBuffer);
                } else {
                    ByteBuffer newBuffer = buf.slice();
                    buf.clear();
                    buf.put(newBuffer);
                }
            }
            this.readBufferMark = 0;
        }
    }

    private void readFrame(ByteBuffer buf, boolean running) {
        int currentReadFrameSize = this.currentReadFrameSize;
        //如果说小于当前要读的数据, 则说明数据不全, 则等待下次一起读
        if (buf.remaining() < currentReadFrameSize) {
            return;
        }
        int limit = buf.limit();
        buf.limit(buf.position() + currentReadFrameSize);
        ReadFrame f = new ReadFrame();
        RpcPbCallback pbCallback = this.workerParams.getCallback();
        pbCallback.setFrame(f);
        //解析数据
        new PbParser().parse(buf, pbCallback);

        int pos = buf.position();
        //重新设置了limit,pos,读取body数据
        buf.position(pbCallback.getBodyStart());
        buf.limit(pbCallback.getBodyLimit());

        try {
            int type = f.getFrameType();
            if (FrameType.TYPE_REQ == type) {
                processIncomingRequest(buf, f, running, currentReadFrameSize);
            } else if (FrameType.TYPE_RESP == type) {
                //对于客户端的读
                processIncomingResponse(buf, f);
            } else {
                log.warn("bad frame type: {}, {}", type, channel);
            }
        } finally {
            this.readBufferMark = pos;
            this.currentReadFrameSize = -1;
            //设置回本次数据读取完成后的状态
            buf.limit(limit);
            buf.position(pos);
        }
    }

    private void processIncomingResponse(ByteBuffer buf, ReadFrame resp) {
        WriteData wo = this.workerParams.getPendingRequests().remove(BitUtil.toLong(channelIndexInWorker, resp.getSeq()));
        if (wo == null) {
            log.debug("pending request not found. channel={}, resp={}", channel, resp);
            return;
        }
        nioStatus.getRequestSemaphore().release();
        WriteFrame req = wo.getData();
        if (resp.getCommand() != req.getCommand()) {
            wo.getFuture().completeExceptionally(new NetException("command not match"));
            return;
        }
        Decoder decoder = wo.getDecoder();
        if (decoder.decodeInIoThread()) {
            try {
                Object body = decoder.decode(buf);
                resp.setBody(body);
            } catch (Throwable e) {
                if (log.isDebugEnabled()) {
                    log.debug("decode fail. {} {}", channel, e.toString());
                }
                wo.getFuture().completeExceptionally(e);
            }
        } else {
            ByteBuffer bodyBuffer = ByteBuffer.allocate(buf.remaining());
            bodyBuffer.put(buf);
            bodyBuffer.flip();
            resp.setBody(bodyBuffer);
        }
        wo.getFuture().complete(resp);
    }

    private void processIncomingRequest(ByteBuffer buf, ReadFrame req, boolean running, int currentReadFrameSize) {
        if (!running) {
            writeErrorInIoThread(req, CmdCodes.STOPPING, null);
            return;
        }
        NioStatus nioStatus = this.nioStatus;
        ReqProcessor p = nioStatus.getProcessors().get(req.getCommand());
        if (p == null) {
            writeErrorInIoThread(req, CmdCodes.COMMAND_NOT_SUPPORT, null);
            return;
        }
        Decoder decoder = p.getDecoder();
        if (nioStatus.getBizExecutor() == null || p.runInIoThread()) {
            if (!fillBody(buf, req, decoder)) {
                return;
            }
            WriteFrame resp;
            try {
                resp = p.process(req, this);
            } catch (Exception e) {
                writeErrorInIoThread(req, CmdCodes.BIZ_ERROR, e.toString());
                return;
            }
            if (resp != null) {
                resp.setCommand(req.getCommand());
                resp.setFrameType(FrameType.TYPE_RESP);
                resp.setSeq(req.getSeq());
                subQueue.enqueue(resp);
            } else {
                writeErrorInIoThread(req, CmdCodes.BIZ_ERROR, "processor return null response");
            }
            return;
        }
        //是在io线程里解码还是在自定义的线程里解码
        if (decoder.decodeInIoThread()) {
            if (!(fillBody(buf, req, decoder))) {
                return;
            }
        } else {
            ByteBuffer body = ByteBuffer.allocate(buf.remaining());
            body.put(buf);
            body.flip();
            req.setBody(body);
        }
        AtomicInteger bytes = nioStatus.getInReqBytes();
        int afterAddBytes = bytes.addAndGet(currentReadFrameSize);
        if (afterAddBytes < nioConfig.getMaxInBytes()) {
            try {
                nioStatus.getBizExecutor().submit(() -> {
                    try {
                        processIncomingRequestInBizThreadPool(req, p, decoder);
                    } finally {
                        bytes.addAndGet(-currentReadFrameSize);
                    }
                });
            } catch (RejectedExecutionException e) {
                //线程池处理不过来,数据量过大
                writeErrorInIoThread(req, CmdCodes.FLOW_CONTROL,
                        "max incoming request: " + nioConfig.getMaxInRequests());
                bytes.addAndGet(-currentReadFrameSize);
            }
        } else {
            writeErrorInIoThread(req, CmdCodes.FLOW_CONTROL,
                    "max incoming request bytes: " + nioConfig.getMaxInBytes());
            bytes.addAndGet(-currentReadFrameSize);
        }
    }

    private void processIncomingRequestInBizThreadPool(ReadFrame req, ReqProcessor p, Decoder decoder) {
        WriteFrame resp;
        try {
            //在业务线程中进行解码
            if (!decoder.decodeInIoThread()) {
                ByteBuffer bodyBuffer = (ByteBuffer) req.getBody();
                req.setBody(decoder.decode(bodyBuffer));
            }
            resp = p.process(req, null);
        } catch (Exception e) {
            writeErrorInIoThread(req, CmdCodes.BIZ_ERROR, e.toString());
            return;
        }
        if (resp != null) {
            resp.setCommand(req.getCommand());
            resp.setFrameType(FrameType.TYPE_RESP);
            resp.setSeq(req.getSeq());
            writeRespInBizThreads(resp);
        } else {
            writeErrorInBizThread(req, CmdCodes.BIZ_ERROR, "processor return null response");
        }
    }

    /**
     * 在业务线程中写回消息
     *
     * @param frame 写回数据
     * @author YangXiaoLong
     * 2022/10/27 16:40
     */
    // invoke by other threads
    private void writeRespInBizThreads(WriteFrame frame) {
        WriteData data = new WriteData(this, frame);
        WorkerParams wp = this.workerParams;
        wp.getIoQueue().write(data);
        wp.getWakeupRunnable().run();
    }

    private void writeErrorInBizThread(Frame req, int code, String msg) {
        ByteBufferWriteFrame resp = new ByteBufferWriteFrame();
        resp.setCommand(req.getCommand());
        resp.setFrameType(FrameType.TYPE_RESP);
        resp.setSeq(req.getSeq());
        resp.setRespCode(code);
        resp.setMsg(msg);
        writeRespInBizThreads(resp);
    }

    private boolean fillBody(ByteBuffer buf, ReadFrame req, Decoder decoder) {
        try {
            Object body = decoder.decode(buf);
            req.setBody(body);
            return true;
        } catch (Throwable e) {
            if (log.isDebugEnabled()) {
                log.debug("decode fail. {} {}", channel, e.toString());
            }
            writeErrorInIoThread(req, CmdCodes.BIZ_ERROR, e.toString());
            return false;
        }
    }

    private void writeErrorInIoThread(ReadFrame req, int code, String msg) {
        ByteBufferWriteFrame resp = new ByteBufferWriteFrame();
        resp.setCommand(req.getCommand());
        resp.setFrameType(FrameType.TYPE_RESP);
        resp.setSeq(req.getSeq());
        resp.setRespCode(code);
        resp.setMsg(msg);
        subQueue.enqueue(resp);
    }


    public ByteBuffer createNewBuffer() {
        // TODO: 2022/9/18 (直接内存 + 堆内存) > 总的内存,docker环境下杀死该进程
        //当一个对象仅仅被weak reference指向, 而没有任何其他strong reference指向的时候, 如果GC运行, 那么这个对象就会被回收。
        // 如果存在强引用同时与之关联，则进行垃圾回收时也不会回收该对象。
        readBuffer = ByteBuffer.allocateDirect(INIT_BUF_SIZE);
        readBufferCache = new WeakReference<>(readBuffer);
        return readBuffer;
    }

    public void close() {
        if (closed) {
            return;
        }
        this.closed = true;
        if (peer != null && peer.getSlChannel() == this) {
            peer.setSlChannel(null);
        }
    }

    public int getAndIncSeq() {
        return this.seq++;
    }

    public IoSubQueue getSubQueue() {
        return subQueue;
    }

    public SocketChannel getChannel() {
        return channel;
    }

    public boolean isClosed() {
        return closed;
    }

    public Peer getPeer() {
        return peer;
    }

    public void setPeer(Peer peer) {
        this.peer = peer;
    }

    public int getChannelIndexInWorker() {
        return channelIndexInWorker;
    }
}
